package gradle.java.demo;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

@Slf4j
public class MsgProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

    private Flow.Subscription subscription;

    private int index = 1;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        log.info("MsgProcessorReceiveMsg={}", item);
        this.submit(item + "-" + index);
        index ++;
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        log.error("error", throwable);
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {
        log.info("MsgProcessorCompleted!");
        this.close();
    }
}
